package com.imooc.coupon.service.impl;

import com.alibaba.fastjson.JSON;
import com.imooc.coupon.constant.Constant;
import com.imooc.coupon.constant.CouponStatus;
import com.imooc.coupon.dao.CouponDao;
import com.imooc.coupon.entity.Coupon;
import com.imooc.coupon.service.IKafkaService;
import com.imooc.coupon.vo.CouponKaFkaMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Optional;

/**
 * @Date 2021/11/10 18:52
 * @Author wangshbv
 * @Description 将cache中coupon的状态同步到DB中
 */
@Slf4j
@Service
public class KafkaServiceImpl implements IKafkaService {

    @Autowired
    private CouponDao couponDao;

    @Override
    @KafkaListener(topics = {Constant.TOPIC}, groupId = "imooc-coupon-1")
    public void consumeCouponKafkaMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            CouponKaFkaMessage kaFkaMessage = JSON.parseObject(message.toString(), CouponKaFkaMessage.class);
            log.info("receive CouponKaFkaMessage:{}", message.toString());
            CouponStatus status = CouponStatus.of(kaFkaMessage.getStatus());
            switch (status) {
                case USEABLE:

                    break;
                case USED:
                    processUsedCoupons(kaFkaMessage, status);
                    break;
                case EXPIRED:
                    processExpiredCoupons(kaFkaMessage, status);
                    break;

            }
        }
    }

    /**
     * 处理已经使用的优惠券
     *
     * @param kaFkaMessage
     * @param status
     */
    private void processUsedCoupons(CouponKaFkaMessage kaFkaMessage, CouponStatus status) {
        //TODO 给用户发送邮件
        processCouponByStatus(kaFkaMessage, status);
    }

    /**
     * 处理过期的优惠券
     *
     * @param kaFkaMessage
     * @param status
     */
    private void processExpiredCoupons(CouponKaFkaMessage kaFkaMessage, CouponStatus status) {
        processCouponByStatus(kaFkaMessage, status);
    }

    /**
     * 根据状态处理优惠劵的信息
     *
     * @param kaFkaMessage
     */
    private void processCouponByStatus(CouponKaFkaMessage kaFkaMessage, CouponStatus status) {
        List<Coupon> coupons = couponDao.findAllById(kaFkaMessage.getIds());
        if (CollectionUtils.isEmpty(coupons) || coupons.size() != kaFkaMessage.getIds().size()) {
            log.error("can not find right coupon info:{}", JSON.toJSONString(kaFkaMessage));
            //todo 发送邮件
            return;
        }
        coupons.forEach(c -> c.setStatus(status));
        log.info("CouponKafkaMessage op coupon count:{}", couponDao.saveAll(coupons).size());

    }
}
